/*-------------------------------------------------------------------------
 *
 * shm_toc.c
 *	  共享内存段目录
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * src/backend/storage/ipc/shm_toc.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "port/atomics.h"
#include "storage/shm_toc.h"
#include "storage/spin.h"

typedef struct shm_toc_entry
{
	uint64		key;			/* 任意标识符 */
	Size		offset;			/* 从 TOC 开始的偏移量，以字节为单位 */
} shm_toc_entry;

struct shm_toc
{
	uint64		toc_magic;		/* 识别此TOC的魔术数字 */
	slock_t		toc_mutex;		/* 用于互斥的自旋锁 */
	Size		toc_total_bytes;	/* 此TOC管理的字节数 */
	Size		toc_allocated_bytes;	/* 已分配的管理字节数 */
	uint32		toc_nentry;		/* TOC中的条目数量 */
	shm_toc_entry toc_entry[FLEXIBLE_ARRAY_MEMBER];
};

/*
 * 使用目录表初始化一块共享内存区域。
 */
shm_toc *
shm_toc_create(uint64 fc_magic, void *fc_address, Size fc_nbytes)
{
	shm_toc    *fc_toc = (shm_toc *) fc_address;

	Assert(fc_nbytes > offsetof(shm_toc, toc_entry));
	fc_toc->toc_magic = fc_magic;
	SpinLockInit(&fc_toc->toc_mutex);

	/*
	 * shm_toc_allocate()中的对齐代码假设起始值是缓冲区对齐的。
	 */
	fc_toc->toc_total_bytes = BUFFERALIGN_DOWN(fc_nbytes);
	fc_toc->toc_allocated_bytes = 0;
	fc_toc->toc_nentry = 0;

	return fc_toc;
}

/*
 * 附加到现有的目录表。如果在目标地址找到的魔术数字
 * 不符合我们的预期，则返回NULL。
 */
shm_toc *
shm_toc_attach(uint64 fc_magic, void *fc_address)
{
	shm_toc    *fc_toc = (shm_toc *) fc_address;

	if (fc_toc->toc_magic != fc_magic)
		return NULL;

	Assert(fc_toc->toc_total_bytes >= fc_toc->toc_allocated_bytes);
	Assert(fc_toc->toc_total_bytes > offsetof(shm_toc, toc_entry));

	return fc_toc;
}

/*
 * 从一个由目录表管理的段中分配共享内存。
 *
 * 这不是一个完整的分配器；没有释放内存的方法。它
 * 只是将一个单一的物理共享内存段分割成可能用于不同目的的逻辑
 * 块的一种方式。
 *
 * 我们是从段的末尾向后分配，以便TOC条目
 * 可以从段的起始处向前增长。
 */
void * shm_toc_allocate(shm_toc *fc_toc, Size fc_nbytes)
{
	volatile shm_toc *fc_vtoc = fc_toc;
	Size		fc_total_bytes;
	Size		fc_allocated_bytes;
	Size		fc_nentry;
	Size		fc_toc_bytes;

	/*
	 * 确保请求对齐良好。 XXX: MAXALIGN不够，
	 * 因为原子操作可能需要更宽的对齐。我们没有
	 * 正确定义原子操作安全所需的最小值，但
	 * BUFFERALIGN应该足够。
	 */
	fc_nbytes = BUFFERALIGN(fc_nbytes);

	SpinLockAcquire(&fc_toc->toc_mutex);

	fc_total_bytes = fc_vtoc->toc_total_bytes;
	fc_allocated_bytes = fc_vtoc->toc_allocated_bytes;
	fc_nentry = fc_vtoc->toc_nentry;
	fc_toc_bytes = offsetof(shm_toc, toc_entry) + fc_nentry * sizeof(shm_toc_entry)
		+ fc_allocated_bytes;

	/* 检查内存耗尽和溢出。 */
	if (fc_toc_bytes + fc_nbytes > fc_total_bytes || fc_toc_bytes + fc_nbytes < fc_toc_bytes)
	{
		SpinLockRelease(&fc_toc->toc_mutex);
		ereport(ERROR,
				(errcode(ERRCODE_OUT_OF_MEMORY),
				 errmsg("out of shared memory")));
	}
	fc_vtoc->toc_allocated_bytes += fc_nbytes;

	SpinLockRelease(&fc_toc->toc_mutex);

	return ((char *) fc_toc) + (fc_total_bytes - fc_allocated_bytes - fc_nbytes);
}

/*
 * 返回仍可分配的字节数。
 */
Size shm_toc_freespace(shm_toc *fc_toc)
{
	volatile shm_toc *fc_vtoc = fc_toc;
	Size		fc_total_bytes;
	Size		fc_allocated_bytes;
	Size		fc_nentry;
	Size		fc_toc_bytes;

	SpinLockAcquire(&fc_toc->toc_mutex);
	fc_total_bytes = fc_vtoc->toc_total_bytes;
	fc_allocated_bytes = fc_vtoc->toc_allocated_bytes;
	fc_nentry = fc_vtoc->toc_nentry;
	SpinLockRelease(&fc_toc->toc_mutex);

	fc_toc_bytes = offsetof(shm_toc, toc_entry) + fc_nentry * sizeof(shm_toc_entry);
	Assert(fc_allocated_bytes + BUFFERALIGN(fc_toc_bytes) <= fc_total_bytes);
	return fc_total_bytes - (fc_allocated_bytes + BUFFERALIGN(fc_toc_bytes));
}

/*
 * 插入一个TOC条目。
 *
 * 这里的想法是，设置共享内存段的进程将
 * 使用此功能注册段内数据结构的地址。每个数据结构
 * 将使用一个64位密钥进行标识，假定这是一个众所周知或可发现的整数。
 * 其他访问共享内存段的进程可以传递相同的密钥
 * 到shm_toc_lookup()以发现那些数据结构的地址。
 *
 * 由于共享内存段可能在不同的后端中以不同的地址映射，
 * 我们存储的是相对指针而非绝对指针。
 *
 * 这在大量密钥的情况下不易扩展。希望这不是
 * 必要的；如果证明是的话，我们可能需要在这里提供一个更复杂的
 * 数据结构。但这里的真正想法只是给映射
 * 动态共享内存的人提供找到他们需要的最少数量的
 * 指针的能力。如果你在TOC中存储了很多东西，
 * 你就是错误的。
 */
void shm_toc_insert(shm_toc *fc_toc, uint64 fc_key, void *fc_address)
{
	volatile shm_toc *fc_vtoc = fc_toc;
	Size		fc_total_bytes;
	Size		fc_allocated_bytes;
	Size		fc_nentry;
	Size		fc_toc_bytes;
	Size		fc_offset;

	/* 相对化指针。 */
	Assert(fc_address > (void *) fc_toc);
	fc_offset = ((char *) fc_address) - (char *) fc_toc;

	SpinLockAcquire(&fc_toc->toc_mutex);

	fc_total_bytes = fc_vtoc->toc_total_bytes;
	fc_allocated_bytes = fc_vtoc->toc_allocated_bytes;
	fc_nentry = fc_vtoc->toc_nentry;
	fc_toc_bytes = offsetof(shm_toc, toc_entry) + fc_nentry * sizeof(shm_toc_entry)
		+ fc_allocated_bytes;

	/* 检查内存耗尽和溢出。 */
	if (fc_toc_bytes + sizeof(shm_toc_entry) > fc_total_bytes ||
		fc_toc_bytes + sizeof(shm_toc_entry) < fc_toc_bytes ||
		fc_nentry >= PG_UINT32_MAX)
	{
		SpinLockRelease(&fc_toc->toc_mutex);
		ereport(ERROR,
				(errcode(ERRCODE_OUT_OF_MEMORY),
				 errmsg("out of shared memory")));
	}

	Assert(fc_offset < fc_total_bytes);
	fc_vtoc->toc_entry[fc_nentry].key = fc_key;
	fc_vtoc->toc_entry[fc_nentry].offset = fc_offset;

	/*
	 * 在填写条目后和更新条目数之前放置一个写屏障，
	 * 可以安全地无锁读取TOC。
	 */
	pg_write_barrier();

	fc_vtoc->toc_nentry++;

	SpinLockRelease(&fc_toc->toc_mutex);
}

/*
 * 查找一个TOC条目。
 *
 * 如果未找到密钥，且noError为真，则返回NULL，
 * 否则抛出elog(ERROR)。
 *
 * 与本文件中的其他函数不同，此操作不获取锁；
 * 它只使用屏障。即使它确实获取了锁，可能也不会对并发性造成太大影响，
 * 但由于一组工作进程可能会在差不多同一时间从同一个TOC中
 * 读取一系列条目，因此避免锁似乎有其价值。
 */
void * shm_toc_lookup(shm_toc *fc_toc, uint64 fc_key, bool fc_noError)
{
	uint32		fc_nentry;
	uint32		fc_i;

	/*
	 * 在我们检查任何条目之前读取条目数量。我们假设
	 * 读取一个uint32是原子的。
	 */
	fc_nentry = fc_toc->toc_nentry;
	pg_read_barrier();

	/* 现在搜索匹配的条目。 */
	for (fc_i = 0; fc_i < fc_nentry; ++fc_i)
	{
		if (fc_toc->toc_entry[fc_i].key == fc_key)
			return ((char *) fc_toc) + fc_toc->toc_entry[fc_i].offset;
	}

	/* 未找到匹配的条目。 */
	if (!fc_noError)
		elog(ERROR, "could not find key " UINT64_FORMAT " in shm TOC at %p",
			 fc_key, fc_toc);
	return NULL;
}

/*
 * 估计存储 TOC 及其依赖数据结构所需的共享内存量。
 */
Size shm_toc_estimate(shm_toc_estimator *fc_e)
{
	Size		fc_sz;

	fc_sz = offsetof(shm_toc, toc_entry);
	fc_sz = add_size(fc_sz, mul_size(fc_e->number_of_keys, sizeof(shm_toc_entry)));
	fc_sz = add_size(fc_sz, fc_e->space_for_chunks);

	return BUFFERALIGN(fc_sz);
}
